#include <carbon/core/error.h>
#include <carbon/core/optional.h>
#include <carbon/core/thread.h>
#include <carbon/core/block_queue.h>
#include <carbon/core/queued_thread_pool.h>
#include <unordered_map>
#include <vector>
#include <thread>
#include <functional>
#include <string>
#include <mutex>
#include <sstream>
#include <iostream>
#include <deque>

#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <netinet/in.h>
#include <netinet/ip.h> 

#include <string.h>
#include <signal.h>
#include <stdio.h>
#include <fcntl.h>

#ifndef __linux__
#error "not linux"
#endif 

using carbon::fd_t;
using carbon::socket_t;
using carbon::Error;
using carbon::Optional;
using carbon::Thread;
using carbon::RunnableFn;
using carbon::QueuedThreadPool;

#define tlogf(...) do { \
    fprintf(stderr, "[%lx] ", static_cast<int64_t>(std::hash<std::thread::id>{}(std::this_thread::get_id())));    \
    fprintf(stderr, __VA_ARGS__); \
    fprintf(stderr, "\n");  \
} while(0)

template<typename T>
class Result {
public:
    Result(int code, const char * msg, const char * file, int line) : m_error(file, line, code, std::string(msg)) {}
    Result(T && value) : m_value(std::move(value)) {}
    Result(const T & value) : m_value(value) {}

    bool success() const { return !m_error; }

    T & value() & { return m_value.get(); }
    const T & value() const & { return m_value.get(); }
    
public:
    static Result ok(T && t) { return Result(std::move(t)); }
    static Result ok(const T & t) { return Result(t); }
    static Result error(int code, const char * msg, const char * file, int line) { return Result(code, msg, file, line); }

private:
    Optional<T> m_value;
    Error       m_error;
}; // Result

template<>
class Result<void> {
public:
    Result() {}
    Result(int ec, const char * msg, const char * file, int line) : m_error(file, line, ec, std::string(msg)) {}

    bool success() const { return !m_error; }

public:
    static Result ok() { return Result(); }
    static Result error(int ec, const char * msg, const char * file, int line) { return Result(ec, msg, file, line); }
    static Result error(int ec, const char * ename, const char * msg, const char * file, int line); // 指定error code名称

private:
    Error   m_error;
};

class SocketProtocol final {
public:
    static SocketProtocol tcp4() { return SocketProtocol{AF_INET, SOCK_STREAM, 0}; }

protected:
    SocketProtocol(int af, int type, int proto) : m_domain(af), m_type(type), m_proto(proto) {}

public:
    int domain() const  { return m_domain; }
    int type() const    { return m_type; }
    int proto() const   { return m_proto; }

private:
    int m_domain    { AF_UNSPEC };
    int m_type  {0};
    int m_proto {0};
};

class SocketAddress {
public:
    static const socklen_t InitCapacity = 128;

    static void initInet(SocketAddress & addr, int af, int port) {
        if ( af == AF_INET) {
            sockaddr_in * p = (sockaddr_in *)addr.data();
            p->sin_family = AF_INET;
            p->sin_addr.s_addr = 0;
            p->sin_port = htons(port);
        }
        else if ( af == AF_INET6 ) {
            sockaddr_in6 * p = (sockaddr_in6 *)addr.data();
            memset(p, 0, sizeof(sockaddr_in6));
            p->sin6_family = AF_INET6;
            p->sin6_port = htons(port);
        }
        else {
            abort();        // 无效分支
        }
    }

public:
    SocketAddress() { this->data()->sa_family = AF_UNSPEC; }
    struct sockaddr * data() { return (sockaddr*)m_data;}
    const struct sockaddr * data() const { return (const sockaddr*)m_data;}

    socklen_t   capacity() const { return InitCapacity; }      
    socklen_t   size() const { return m_size; } 
private:
    char        m_data[InitCapacity];
    socklen_t   m_size {InitCapacity};
};

class Socket {
    CARBON_NONCOPYABLE(Socket)
protected:
    Socket() noexcept {}
    explicit Socket(socket_t sock) noexcept : m_sock(sock) {}
    Socket(const SocketProtocol & proto) { this->create(proto); }
    Socket(Socket && other) noexcept : m_sock(other.m_sock) { other.m_sock = CARBON_INVALID_SOCKET; }

    Socket & operator=(Socket &&other) noexcept {
        if ( this == &other ) return *this;
        assert(m_sock == CARBON_INVALID_SOCKET);    // 当前socket未关闭不能Move

        this->m_sock = other.m_sock;
        other.m_sock = CARBON_INVALID_SOCKET;
        return *this;
    }

    Result<void> create(const SocketProtocol& proto) {
        int s = ::socket(proto.domain(), proto.type() | SOCK_CLOEXEC, proto.proto());
        if ( s == CARBON_INVALID_SOCKET) return Result<void>::error(errno, "socket() error", CARBON_FILE_LINE);

        m_sock = s;
        return Result<void>::ok();
    } 

public:
    virtual ~Socket() { 
        assert(m_sock == CARBON_INVALID_SOCKET);    // 必须显式关闭
    }
    
    bool     valid() const  { return m_sock != CARBON_INVALID_SOCKET; }
    socket_t handle() const { return m_sock; }


    void close() {
        this->doClose();
        if ( m_sock != CARBON_INVALID_SOCKET) {
            ::close(m_sock);
            m_sock = CARBON_INVALID_SOCKET;
        }
    }

    socket_t release() {
        this->doClose();
        socket_t fd = m_sock;
        m_sock = CARBON_INVALID_SOCKET;
        return fd;
    }
    
    /// 获取当前socket是否非阻塞
    Result<bool> nonblocking() const {
        int r = ::fcntl(m_sock, F_GETFL);
        if ( r < 0 ) return Result<bool>::error(errno, "fcntl(F_GETFL) error", CARBON_FILE_LINE);
    
        bool nonblock = (r & O_NONBLOCK) != 0;
        return Result<bool>::ok(nonblock);
    }

    // 设置当前socket是否非阻塞
    Result<void> nonblocking(bool value) {
        int r = ::fcntl(m_sock, F_GETFL);
        if ( r < 0 ) return Result<void>::error(errno, "fcntl(F_GETFL) error", CARBON_FILE_LINE);

        if ( value ) r |= O_NONBLOCK;
        else r &= ~O_NONBLOCK;

        r = ::fcntl(m_sock, F_SETFL, r);
        if ( r < 0 ) return Result<void>::error(errno, "fcntl(F_SETFL) error", CARBON_FILE_LINE);

        return Result<void>::ok();
    }

    /// 获取socket domian
    Result<int> domain() const {
        int domain;
        socklen_t len = sizeof(domain);
        int r = ::getsockopt(m_sock, SOL_SOCKET, SO_DOMAIN, &domain, &len);
        if ( r == 0 ) return domain;
        else return Result<int>::error(errno, "getsockopt(SOL_SOCKET, SO_DOMAIN)", CARBON_FILE_LINE);
    }

protected:
    virtual void doClose() {}

private:
    socket_t  m_sock { CARBON_INVALID_SOCKET };
}; // Socket


class StreamSocket : public Socket {
public:
    StreamSocket(socket_t sock) : Socket(sock) {}
};

class ServerSocket : public Socket {
public:
    ServerSocket() {}

    ServerSocket(const SocketProtocol &proto, int port) : Socket(proto) {
        if ( this->valid() ) this->open(port);
    }

    ServerSocket(socket_t sock) : Socket(sock) {}

    ServerSocket(ServerSocket && other) : Socket(std::move(other)) {}

    virtual ~ServerSocket() {}

    ServerSocket & operator=(ServerSocket && other) {
        this->Socket::operator=(std::move(other));
        return *this;
    }

    Result<void> open(int port) {
        if ( !this->valid() ) { // socket未创建则先创建
            auto r0 = this->create(SocketProtocol::tcp4());
            if ( !r0.success() ) return r0; // 创建失败则返回
        }

        this->reuseAddr(true);

        Result<int> domain = this->domain();
        assert(domain.success());
        SocketAddress addr;
        SocketAddress::initInet(addr, domain.value(), port);

        int r1 = ::bind(this->handle(), addr.data(), addr.size());
        if ( r1 == -1 ) {
            this->close();  // 关闭回退
            return Result<void>::error(errno, "bind() error", CARBON_FILE_LINE);
        }

        int r2 = ::listen(this->handle(), 100);
        if ( r2 == -1 ) {
            this->close();  // 关闭回退
            return Result<void>::error(errno, "listen() error", CARBON_FILE_LINE);
        }

        return Result<void>::ok();
    }

    Result<StreamSocket> accept() {
        assert(this->valid());
        SocketAddress addr;
        socklen_t len = addr.capacity();
        int r = ::accept(this->handle(), addr.data(), &len);
        if ( r == -1 ) return Result<StreamSocket>::error(errno, "accept() error", CARBON_FILE_LINE);

        return Result<StreamSocket>::ok(StreamSocket(r));
    }

    Result<void> reuseAddr(bool reuse) {
        int value = value ? 1 : 0; 
        int r = ::setsockopt(this->handle(), SOL_SOCKET, SO_REUSEADDR, &value, sizeof(value));
        if ( r == -1 ) return Result<void>::error(errno, "setsockopt(SO_REUSEADDR) error", CARBON_FILE_LINE);
        return Result<void>::ok();
    }

}; // ServerSocket



class DatagramSocket : public Socket {

};


class Selector {
public:
    using SelectedFn = std::function<void(int ops)>;

    static const int OpNone     = 0;    ///< 无
    static const int OpRead     = 1;    ///< 可读
    static const int OpWrite    = 2;    ///< 可写
    static const int OpTimeout  = 4;    ///< 超时   
    static const int OpError    = 8;    ///< socket发生错误
    static const int OpOneshot  = 16;   ///< 针对一个socket的事件仅触发一次，仅在请求时设置
    static const int OpRemove   = 32;   ///< socket从selector中移除

    struct Request {
        fd_t        fd;     // fd
        int         ops;    // request ops
        SelectedFn  cb;     // event callback
    };
    using RequestMap = std::unordered_map<fd_t, Request>;

    struct Selection {
        fd_t fd;
        int  ops;
    };
    using SelectionVec = std::vector<Selection>;
protected:
    Selector() {}
    Selector(Selector &&other);

public:
    virtual ~Selector() {}

    Result<void> add(fd_t fd, int ops, SelectedFn cb) {
        Request request {fd, ops, std::move(cb)};

        auto r1 = this->doAdd(request);
        if ( !r1.success() ) return r1;

        auto r2 = m_requests.insert(std::make_pair(fd, std::move(request)));
        assert(r2.second);
        return r1;
    }

    Result<void> modify(fd_t fd, int ops) {
        auto it = m_requests.find(fd);
        assert(it != m_requests.end());
        it->second.ops = ops;

        auto r1 = this->doModify(it->second);
        return r1;
    }

    void remove(fd_t fd) {
        this->doRemove(fd);
        auto it = m_requests.find(fd);
        if ( it == m_requests.end() ) return ;

        if ( it->second.cb ) it->second.cb(OpRemove);
        m_requests.erase(fd);
    }

    Result<int>  select() {
        std::chrono::milliseconds timeout{3600000}; // 默认1小时阻塞代替无限时间阻塞
        return this->select(timeout);
    }

    Result<int> select(std::chrono::milliseconds timeout) {
        tlogf("Selector::select begin");
        auto r1 = this->doSelect(m_selections, timeout);
        if ( !r1.success() || r1.value() == 0 ) return r1;

        for ( auto & sel : m_selections ) {
            auto it = m_requests.find(sel.fd);
            assert(it != m_requests.end());
            if(it->second.cb) it->second.cb(sel.ops);   // notifier可能不会设置cb
        }
        return r1;    
    }

protected:
    virtual Result<void> doAdd(const Request &request) = 0;
    virtual Result<void> doModify(const Request &request) = 0;
    virtual void         doRemove(fd_t fd) = 0;
    virtual Result<int>  doSelect(SelectionVec & selections, std::chrono::milliseconds timeout) = 0;
protected:
    RequestMap      m_requests;
    SelectionVec    m_selections;
}; // Selector

class EPollSelector : public Selector {
    CARBON_NONCOPYABLE(EPollSelector)
public:
    EPollSelector() {
        m_epoll_fd = ::epoll_create1(EPOLL_CLOEXEC);
        assert(m_epoll_fd != CARBON_INVALID_FD);
    }

    virtual ~EPollSelector() {
        if ( m_epoll_fd != CARBON_INVALID_FD ) {
            ::close(m_epoll_fd);
            m_epoll_fd = CARBON_INVALID_FD;
        }
    }

public:
    virtual Result<void> doAdd(const Request &request) override final {
        struct epoll_event event;
        event.data.fd = request.fd;
        event.events = 0;
        if ( request.ops & OpRead ) event.events |= EPOLLIN;
        if ( request.ops & OpWrite ) event.events |= EPOLLOUT;
        if ( request.ops & OpOneshot) event.events |= EPOLLONESHOT;

        int r = ::epoll_ctl(m_epoll_fd, EPOLL_CTL_ADD, request.fd, &event);
        if ( r == -1 ) return Result<void>::error(errno, "epoll_ctl(EPOLL_CTL_ADD) error", CARBON_FILE_LINE);

        m_events.resize(m_events.size() + 1);
        tlogf("epoll add fd: %d, ops: %d", request.fd, request.ops);
        return Result<void>::ok();
    }

    virtual Result<void> doModify(const Request &request) override final {
        struct epoll_event event;
        event.data.fd = request.fd;
        event.events = 0;
        if ( request.ops & OpRead ) event.events |= EPOLLIN;
        if ( request.ops & OpWrite ) event.events |= EPOLLOUT;
        if ( request.ops & OpOneshot) event.events |= EPOLLONESHOT;

        tlogf("epoll modify fd: %d, ops: %d, events: %x", request.fd, request.ops, event.events);

        int r = ::epoll_ctl(m_epoll_fd, EPOLL_CTL_MOD, request.fd, &event);
        if ( r == -1 ) return Result<void>::error(errno, "epoll_ctl(EPOLL_CTL_ADD) error", CARBON_FILE_LINE);
        return Result<void>::ok();
    }

    virtual void doRemove(fd_t fd) override final {
        int r = ::epoll_ctl(m_epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
        if ( r == 0 && !m_events.empty()) m_events.resize(m_events.size() - 1);
    }

    virtual Result<int> doSelect(SelectionVec & selections, std::chrono::milliseconds timeout) override final {
        auto now = std::chrono::steady_clock::now();
        auto exp = now + timeout;
            
        while( now < exp ) { 
            auto t1 = exp - now;
            auto t2 = std::chrono::duration_cast<std::chrono::milliseconds>(t1);

            int r = ::epoll_wait(m_epoll_fd, m_events.data(), m_events.size(), (int)t2.count());
            if ( r >= 0 ) {
                selections.resize(r);
                for (int i = 0; i < r; ++i ) {
                    auto & sel = selections[i];
                    auto & evt = m_events[i];
                    sel.fd = evt.data.fd;
                    sel.ops = 0;
                    if (evt.events & EPOLLIN ) sel.ops |= Selector::OpRead;
                    if (evt.events & EPOLLOUT) sel.ops |= Selector::OpWrite;
                    if (evt.events & EPOLLERR) sel.ops |= Selector::OpError;
                    tlogf("EPollSelector::doSelect fd: %d, ops: %d, events: %d", sel.fd, sel.ops, evt.events);
                }
                return Result<int>::ok(r);
            }

            // 错误处理
            int err = errno;
            if ( err == EINTR ) {   // 中断，重试
                now = std::chrono::steady_clock::now();
                continue;   
            }
            else {
                return Result<int>::error(err, "epoll_wait() error", CARBON_FILE_LINE);
            }
        } // while

        return Result<int>::ok(0);  // 超时
    }

private:
    int                         m_epoll_fd {CARBON_INVALID_FD};
    std::vector<epoll_event>    m_events;  // event results
}; // EPollSelector



using DefaultSelector = EPollSelector;

/// @brief 通知器
class EventFdNotifier final {
public:
    EventFdNotifier() {
        m_eventfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        assert(m_eventfd != CARBON_INVALID_FD);
    }

    ~EventFdNotifier() {
        if ( m_eventfd != CARBON_INVALID_FD ) {
            ::close(m_eventfd);
            m_eventfd = CARBON_INVALID_FD;
        }
    }

    fd_t handle() const { return m_eventfd; }
    void notify() {
        uint64_t one = 1;
        ::write(m_eventfd, &one, sizeof(one));
    }

    void reset() {
        uint64_t one = 1;
        ::read(m_eventfd, &one, sizeof(one));
    }
private:
    int m_eventfd { CARBON_INVALID_FD};
};

using DefaultNotifier = EventFdNotifier;

/// @brief 异步任务通道
class TaskChannel {
public:
    using Queue = std::queue<RunnableFn>;

public:
    TaskChannel() {}    // TODO queue capacity支持

    fd_t handle() const { return m_notifier.handle(); }

    ssize_t send(RunnableFn  task) {
        std::lock_guard<std::mutex> lock(m_lock);
        m_queue.push(std::move(task));
        m_notifier.notify();
        return 1;
    }

    ssize_t receive(RunnableFn tasks[], size_t size) {
        std::lock_guard<std::mutex> lock(m_lock);
        m_notifier.reset();
        ssize_t count = 0;
        while ( !m_queue.empty() && count < size ) {
            tasks[count++] = std::move(m_queue.front());
            m_queue.pop();
        }
        return count;
    }

private:
    DefaultNotifier m_notifier;
    std::mutex      m_lock;
    Queue           m_queue;
};

class SelectorService {
public:
    using SelectedFn = Selector::SelectedFn;
    
public:
    SelectorService() {
        // 注册任务队列，任务队列不需要回调
        auto r = m_selector.add(m_channel.handle(), Selector::OpRead, nullptr);
        assert(r.success());
    }

    Result<void> add(fd_t fd, int ops, SelectedFn cb) {
        auto fn = [this, fd, ops, cb]() {
            this->doAdd(fd, ops, cb);
        };
        auto ok = m_channel.send(fn);
        if ( ok ) return Result<void>::ok();
        else return Result<void>::error(-1, "queue full", CARBON_FILE_LINE);
    }

    Result<void> modify(fd_t fd, int ops) {
        auto fn = [this, fd, ops]() {
            this->doModify(fd, ops);
        };
        auto ok = m_channel.send(fn);
        if ( ok ) return Result<void>::ok();
        else return Result<void>::error(-1, "queue full", CARBON_FILE_LINE);
    }

    Result<void> remove(fd_t fd) {
        auto fn = [this, fd]() {
            this->doRemove(fd);
        };
        auto ok = m_channel.send(fn);
        if ( ok ) return Result<void>::ok();
        else return Result<void>::error(-1, "queue full", CARBON_FILE_LINE);
    }

    Result<void> start() {
        m_thread = Thread([this](){
                this->doRun();
            },
            "SelectorService"
        );
        return Result<void>::ok();
    }

    void stop() {
        m_stop = true;
        m_channel.send([](){}); // 发送空任务唤醒
        m_thread.join();    // 等待线程结束
    }

private:
    void doAdd(fd_t fd, int ops, SelectedFn cb) {
        auto r = m_selector.add(fd, ops, std::move(cb));
        assert( r.success());
    }

    void doModify(fd_t fd, int ops) {
        auto r = m_selector.modify(fd, ops);
        assert( r.success());
    }

    void doRemove(fd_t fd) {
        m_selector.remove(fd);
    }

    void doRun() {
        while ( !m_stop ) {
            this->doTasks();    // 执行队列任务
            this->doSelect();   // 执行select等待 
        }
    }

    void doTasks() {
        RunnableFn tasks[8];
        while ( 1 ) {
            ssize_t n = m_channel.receive(tasks, 8);
            for ( ssize_t i = 0; i < n; ++i ) tasks[i]();
            if ( n < 8) break;
        }
    }

    void doSelect() {
        tlogf("SelectorService::doSelect");
        m_selector.select();
        // TODO 计时器控制超时
    }

private:
    DefaultSelector     m_selector;
    TaskChannel         m_channel;
    volatile bool       m_stop {false};
    Thread              m_thread;       // selector thread
};


class SocketAcceptService {
public:
    using OnAcceptFn = std::function<void(socket_t servsock, std::vector<socket_t> &sockets)>;
public:

    SocketAcceptService(OnAcceptFn cb) : m_threads("acceptsvc", 1), m_onAcceptFn(cb) {}

    Result<void> accept(socket_t servsock) {
        auto fn = [this, servsock](){ this->doAccept(servsock); };
        bool ok = m_threads.execute(fn);
        if ( ok ) return Result<void>::ok();
        else return Result<void>::error(-1, "queue full", CARBON_FILE_LINE);
    }

    Result<void> start() {
        m_threads.start();
        return Result<void>::ok();
    }

    void stop() {
        m_threads.stop();
    }

private:
    void doAccept(socket_t servsock) {
        ServerSocket server_socket(servsock);
        std::vector<socket_t> sockets;
        tlogf("SocketAcceptService::doAccept begin");
        while ( 1 ) {
            Result<StreamSocket> socket = server_socket.accept();
            if ( socket.success() ) {
                if ( sockets.size() == sockets.capacity()) sockets.reserve(sockets.size() + 100);
                sockets.push_back(socket.value().release());
            }
            else break;
        }
        this->m_onAcceptFn(servsock, sockets);
        tlogf("SocketAcceptService::doAccept end");
        server_socket.release();
    }

private:
    OnAcceptFn       m_onAcceptFn;
    QueuedThreadPool m_threads;
}; // SocketAcceptService

template<typename Buf>
class BufferView {
public:
    using ValueType = typename Buf::ValueType;
public:
    BufferView(ValueType * data, size_t size)
    : m_data(data), m_size(size) {}

    ValueType * data() { return m_data; }
    ValueType * data() const { return m_data; }
    size_t      size() const { return m_size; }

private:
    ValueType * m_data;
    size_t      m_size;
};

template<typename Buf>
class BufferViews {
public:
    using ValueType = typename Buf::ValueType;
    using View = BufferView<Buf>;
    using Visitor = std::function<bool(ValueType * data, size_t size)>;

public:
    size_t count() const { return m_views.size(); }

    void push(View &&  view) {
        m_views.push_back(std::move(view));
    }

    void accept(size_t skip_bytes, Visitor visitor) const {
        for ( auto & view : m_views ) {
            if ( view.size() <= skip_bytes ) {
                skip_bytes -= view.size();
            }
            else {
                bool ok = visitor(view.data() + skip_bytes, view.size() - skip_bytes);
                if ( !ok ) break;   // 不用再输出
                skip_bytes = 0; // 全部略过
            }
        }
    }

private:
    std::deque<View> m_views;
};


class Buffer {
protected:
    Buffer() = default;
    Buffer(const Buffer & other) = default;
    Buffer(Buffer && other) 
    : m_capacity(other.m_capacity), m_limit(other.m_limit), m_position(other.m_position), m_mark(other.m_mark) 
    {
        other.m_capacity = other.m_limit = other.m_position = other.m_mark = 0;
    }

    Buffer(size_t capacity) : m_capacity(capacity), m_limit(capacity) {}

    Buffer & operator=(const Buffer &) = delete;

    Buffer & operator=(Buffer && other);
public:
    virtual ~Buffer() { 
        m_capacity = m_limit = m_position = m_mark = 0;
    }

    size_t capacity() const { return m_capacity; }
    size_t limit() const { return m_limit; }
    size_t position() const { return m_position; }
    size_t mark() const { return m_mark; }
    size_t remaining() const { return m_limit - m_position; }

    void clear() {
        m_limit = m_capacity;
        m_position = 0;
        m_mark = 0;
    }

    void position(size_t pos) {
        assert(pos <= m_limit);
        m_position = pos;
        if ( m_mark > pos ) m_mark = 0;
    }

    void mark() { m_mark = m_position; }

    void limit(size_t limit) {
        assert(limit <= m_capacity);
        m_limit = limit;
        if ( m_position > limit ) m_position = limit;
        if ( m_mark > limit ) m_mark = 0;
    }

    void reset() {
        m_position = m_mark;
    }

    void rewind() {
        m_position = 0;
        m_mark = 0;
    }

    void flip() {
        m_limit = m_position;
        m_position = 0;
        m_mark = 0;
    }

protected:
    size_t m_capacity   {0};
    size_t m_limit      {0};
    size_t m_position   {0};
    size_t m_mark       {0};
}; // Buffer

class ByteBuffer final : public Buffer  {
    CARBON_NONCOPYABLE(ByteBuffer)
public:
    using ValueType = char;
    static ByteBuffer allocate(size_t size) {
        return ByteBuffer(new char[size], size);
    }

private:
    ByteBuffer(char * data, size_t capacity) : Buffer(capacity), m_data(data) {}
    
public:
    ByteBuffer() {}
    ByteBuffer(ByteBuffer && other) 
    : Buffer(std::move(other)), m_data(other.m_data)
    {
        other.m_data = nullptr;
    }

    virtual ~ByteBuffer() { 
        if( m_data ) {
            delete[] m_data;
            m_data = nullptr;
        }
    }

    ByteBuffer & operator=(ByteBuffer && other);

    char* data() const { return m_data; }

    BufferView<ByteBuffer> spaceView() const {
        return BufferView<ByteBuffer>(m_data + m_position, m_limit - m_position);
    }

    BufferView<ByteBuffer> dataView() const {
        return BufferView<ByteBuffer>(m_data, m_position);
    }

private:
    char * m_data {nullptr};
}; // ByteBuffer

template<typename Buf>
class Buffers {
public:
    using Visitor = std::function<bool(const Buf & buf)>;
public:
    bool empty() const { return m_buffers.empty(); }

    size_t count() const { return m_buffers.size(); }

    void push(Buf && buf) {
        m_buffers.push_back(std::move(buf));
    }

    void pop () {
        m_buffers.pop_front();
    }

    Buf & front() { return m_buffers.front(); }

    void commit(size_t bytes) {
        for ( auto & buf : m_buffers ) {
            if ( bytes == 0 ) break;    // 提交完成
            size_t remain = buf.remaining();
            if ( remain  == 0 ) continue;
            if ( remain <= bytes ) {
                buf.position(buf.limit());
                bytes -= remain;
            }
            else {
                buf.position(buf.position() + bytes);
                bytes = 0;
            }
        }
    }

    void accept(Visitor visitor) {
        for ( auto & buf : m_buffers ) {
            if ( buf.position() == buf.limit()) continue;
            if ( !visitor(buf) ) break;
        }
    }


private:
    std::deque<Buf> m_buffers;
}; // Buffers


using ByteBufferView = BufferView<ByteBuffer>;
using ByteBufferViews = BufferViews<ByteBuffer>;




class SocketIOService {
public:
    enum class IOStatus {
        Success,
        Error,
        EndOfStream,
    };
    using IOCallback = std::function<void(size_t recvBytes, size_t sendBytes, IOStatus status)>;

public:
    SocketIOService() : m_threads("iosvc", 1) {}

    Result<void> start() { 
        m_threads.start();
        return Result<void>::ok();
    }

    void stop() {
        m_threads.stop();
    }

    Result<void> process(socket_t socket, ByteBufferViews recvBuffers, ByteBufferViews sendBuffers, IOCallback cb) {
        auto fn = [this, socket, sendBuffers= std::move(sendBuffers), recvBuffers = std::move(recvBuffers), cb = std::move(cb)]() {
            IOStatus status = IOStatus::Success;
            size_t recv_bytes = 0, send_bytes = 0;
            if ( status == IOStatus::Success && recvBuffers.count() > 0 ) this->doReceiveSome(socket, recvBuffers, &recv_bytes, &status);
            if ( status == IOStatus::Success && sendBuffers.count() > 0 ) this->doSendSome(socket, sendBuffers, &send_bytes, &status);

            cb(recv_bytes, send_bytes, status);
        };

        bool ok = m_threads.execute(fn);
        if ( ok ) return Result<void>::ok();
        else return Result<void>::error(-1, "queue full", CARBON_FILE_LINE);
    }


private:
    void doSendSome(socket_t socket, const ByteBufferViews & buffers, size_t * bytes, IOStatus * status) {
        size_t total_sent = 0;  // 本次调用总接收字节数
        
        // 设置发送消息对象
        struct iovec  iov[8];
        struct msghdr msg;
        memset(&msg, 0, sizeof(msg));
        msg.msg_iov = iov;
        
        while (1) {
            msg.msg_iovlen = 0;

            // 访问者模式获取数据，输入total_sent表示遍历缓存时，需要跳过的字节数（这部分已经发送）
            buffers.accept(total_sent, [&](const char * data, size_t size) -> bool {
                if (msg.msg_iovlen == 8 ) return false; // 无需再遍历
            
                iov[msg.msg_iovlen].iov_base = (char*)data;
                iov[msg.msg_iovlen].iov_len = size;
                ++msg.msg_iovlen;

                if ( msg.msg_iovlen == 8) return false; // 本次缓存已满
                else return true;
            });

            if ( msg.msg_iovlen == 0 ) {    // 缓存已经收满，无需再接收
                *bytes = total_sent;
                *status = IOStatus::Success;
                return; 
            }

            // 执行消息发送
            ssize_t r = ::sendmsg(socket, &msg, 0);
            if ( r > 0 ) {
                total_sent += r;
                continue;       // 收到，继续接收
            }

            if ( r == 0 ) {     // 发送缓存已满
                *bytes = total_sent;
                *status = IOStatus::Success;
                return ;
            }

            if ( r < 0 ) {
                int ec = errno;
                if ( ec == EAGAIN ) {   // 本次接收完成
                    *bytes = total_sent;
                    *status = IOStatus::Success;
                    return ;
                }
                else if ( ec == EINTR ) {
                    continue;   // 中断重试
                }
                else {  // 接收错误
                    *bytes = total_sent;
                    *status = IOStatus::Error;
                    return ;
                }
            }
        } // while
    } // doSendSome

    void doReceiveSome(socket_t socket, const ByteBufferViews & buffers, size_t * bytes, IOStatus * status) {
        size_t total_received = 0;  // 本次调用总接收字节数
        
        // 设置发送消息对象
        struct iovec  iov[8];
        struct msghdr msg;
        memset(&msg, 0, sizeof(msg));
        msg.msg_iov = iov;
        
        tlogf("SocketIOService::doReceiveSome fd: %d", socket);

        while (1) {
            msg.msg_iovlen = 0;

            // 访问者模式获取空闲缓存空间，输入total_received表示遍历缓存时，需要跳过的字节数（这部分已经收到写入）
            size_t skip_bytes = total_received;
            buffers.accept(total_received, [&iov, &msg, bytes=total_received](char * data, size_t size) -> bool {
                if (msg.msg_iovlen == 8 ) return false; // 无需再遍历

                iov[msg.msg_iovlen].iov_base = data;
                iov[msg.msg_iovlen].iov_len = size;
                ++msg.msg_iovlen;

                if ( msg.msg_iovlen == 8) return false; // 本次缓存已满
                else return true;
            });

            if ( msg.msg_iovlen == 0 ) { // 缓存已经收满，无需再接收
                *bytes = total_received;
                *status = IOStatus::Success;
                break;
            }

            // 执行消息发送
            ssize_t r = ::recvmsg(socket, &msg, 0);
            if ( r > 0 ) {
                total_received += r;
                continue;       // 收到，继续接收
            }

            if ( r == 0 ) { // 对端关闭
                *bytes = total_received;
                *status = IOStatus::EndOfStream;
                return;
            }

            if ( r < 0 ) {
                int ec = errno;
                if ( ec == EAGAIN ) {   // 本次接收完成
                    *bytes = total_received;
                    *status = IOStatus::Success;
                    return ;
                }
                else if ( ec == EINTR ) {
                    continue;   // 中断重试
                }
                else {
                    *bytes = total_received;
                    *status = IOStatus::Error; // 接收错误
                    return ;
                }
            }
        } // while
    } // doReceiveSome

private:
    QueuedThreadPool m_threads;
};


struct Session {
    socket_t               socket {CARBON_INVALID_SOCKET};
    Buffers<ByteBuffer>    readBuffers;    // 接收数据缓冲区
    Buffers<ByteBuffer>    writeBuffers;   // 发送数据缓冲区

    Session(socket_t socket) : socket(socket) {}
};

class EchoServer {
private:
    static constexpr const int ThreadCount = 2;

    using SessionMap = std::unordered_map<socket_t, Session>;
public:
    EchoServer(int port) 
    : m_listen_port(port)
    , m_acceptor([this](socket_t servsock, std::vector<socket_t> &sockets){ this->onAccepted(servsock, sockets); }) {}

    void start();
    void stop();
    
private:
    void run();

    void onAccepted(socket_t servsock, std::vector<socket_t> & sockets) {
        this->doNewSessions(servsock, sockets);
        m_selector.modify(servsock, Selector::OpRead | Selector::OpOneshot);
    }

    void onReceived(Session & session, size_t bytes) {
        session.readBuffers.commit(bytes);
        while ( !session.readBuffers.empty() ) {
            auto & buffer = session.readBuffers.front();
            if ( buffer.position() > 0 ) {      // Echo逻辑，只要读到数据，就执行发送，不需要读满缓存再发
                buffer.flip();
                session.writeBuffers.push(std::move(buffer));
                session.readBuffers.pop();
            }
            else break;
        }
    }

    void onSent(Session & session, size_t bytes) {
        session.writeBuffers.commit(bytes);
        while ( !session.writeBuffers.empty()) {
            auto & buffer = session.writeBuffers.front();
            if ( buffer.remaining() == 0 ) {        // 发送消息则需要全部发完之后才能继续读
                buffer.clear();
                session.readBuffers.push(std::move(buffer));
                session.writeBuffers.pop();
            }
            else break;
        }
    } // onSent

    void doNewSessions(socket_t servsock, std::vector<socket_t> & sockets) {
        std::lock_guard<std::mutex> lock(m_sessions_lock);
        for ( auto & sock : sockets ) {
            auto r1 = m_sessions.insert(std::make_pair(sock, Session(sock)));
            assert(r1.second);
            Session & session = r1.first->second;
            ByteBuffer read_buffer = ByteBuffer::allocate(1024);
            session.readBuffers.push(std::move(read_buffer));
            
            // 设置为非阻塞
            StreamSocket socket(sock);
            auto r2 = socket.nonblocking(true);
            assert(r2.success());
            socket.release();

            tlogf("EchoServer::doNewSession sock: %d", sock);

            auto fn = [this, sock, &session](int ops) {
                if ( ops & Selector::OpRemove ) {
                    this->doSessionClose(session);  // selector中移除完成，可以直接关闭连接
                    return;
                }
                if (ops & Selector::OpError || ops & Selector::OpTimeout ) {
                    this->m_selector.remove(sock);  // 通知selector删除socket
                    return ;
                }
                this->doSessionReadWrite(session, ops);
            };
            auto r3 = m_selector.add(sock, Selector::OpRead | Selector::OpOneshot, std::move(fn));
            assert(r3.success());
        }
    }

    void doSessionReadWrite(Session & session, int ops) {

        tlogf("doSessionReadWrite sock: %d, ops: %d", session.socket, ops);
        auto fn = [this, &session](size_t rbytes, size_t wbytes, SocketIOService::IOStatus status) {
            
            // 判断io是否异常，如果异常，就关闭连接。
            // 当前是io线程的回调，且同一时间不会有其他线程处理这个连接的io，所以可以直接关闭连接
            // 关闭连接的过程，首先将socket从selector中移除，然后再关闭连接。
            // 如果是先关闭连接，再异步通知selector删除socket，中间的时间差可能会因新连接而生成新的同值socket, 导致selector旧的还没删除，新的加入出现重复socket。
            // selector remove之后，会执行最有一次selector回调，此时再做连接实际删除。
            if ( status == SocketIOService::IOStatus::EndOfStream || status == SocketIOService::IOStatus::Error ) {
                m_selector.remove(session.socket);  // 通知selector删除该socket
                return;
            }


            assert(status == SocketIOService::IOStatus::Success);
            
            if ( rbytes > 0 ) this->onReceived(session, rbytes);
            if ( wbytes > 0 ) this->onSent(session, wbytes);

            int ops = Selector::OpOneshot;
            if ( !session.writeBuffers.empty() ) ops |= Selector::OpWrite;
            if ( !session.readBuffers.empty() ) ops |= Selector::OpRead;

            tlogf("doSessionReadWrite-cb sock: %d, ops: %d, rbytes: %ld, wbytes: %ld", session.socket, ops, rbytes, wbytes);

            auto r1 = m_selector.modify(session.socket, ops);
            assert(r1.success());
        };

        ByteBufferViews rviews;
        ByteBufferViews wviews;

        if ( ops & Selector::OpRead ) {
            session.readBuffers.accept([&](const ByteBuffer & buffer) -> bool { 
                ByteBufferView view(buffer.data() + buffer.position(), buffer.remaining());
                rviews.push(std::move(view));
                return true;
            });
        }
        if ( ops & Selector::OpWrite ) {
            session.writeBuffers.accept([&](const ByteBuffer & buffer) -> bool{ 
                ByteBufferView view(buffer.data() + buffer.position(), buffer.remaining());
                wviews.push(std::move(view));
                return true;
            });
        }

        auto r1 = this->m_iosvc.process(session.socket, rviews, wviews, fn);
        assert(r1.success());
    }
    
    void doSessionClose(Session & session) {
        if ( session.socket != CARBON_INVALID_SOCKET) {
            StreamSocket socket(session.socket);
            socket.close();
        }

        std::lock_guard<std::mutex> lock(m_sessions_lock);
        m_sessions.erase(session.socket);
    }

private:
    int                     m_listen_port;
    SelectorService         m_selector;
    SocketAcceptService     m_acceptor;
    SocketIOService         m_iosvc;

    ServerSocket    m_listener;
    std::mutex      m_sessions_lock;
    SessionMap      m_sessions;  
}; // EchoServer

void EchoServer::start()
{
    m_listener = ServerSocket(SocketProtocol::tcp4(), m_listen_port);
    assert(m_listener.valid());

    auto r1 = m_listener.nonblocking(true);
    assert(r1.success() );

    m_iosvc.start();
    m_acceptor.start();
    m_selector.start();

    auto fn = [this](int ops) {
        assert( ops | Selector::OpRead );
        tlogf("EchoServer, accept");
        m_acceptor.accept(m_listener.handle());
    };

    m_selector.add(m_listener.handle(), Selector::OpRead | Selector::OpOneshot, std::move(fn));
}

void EchoServer::stop()
{
    m_selector.stop();
    m_acceptor.stop();
    m_iosvc.stop();

    m_listener.close();
}

volatile bool g_stop = false;

void stop(int sig)
{
    g_stop = true;
}

int main(int argc, char *argv[])
{
    if ( argc < 2 ) {
        printf("Usage: %s <port>\n", argv[0]);
        return 1;
    }

    signal(SIGINT, stop);
    signal(SIGTERM, stop);

    int port = atoi(argv[1]);
    EchoServer server(port);
    server.start();
    
    while (!g_stop) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    server.stop();
    return 0;
}
